#include "Spark/Net/TcpClient.h"
#include "Spark/Net/Impl/Connector.h"
#include "Spark/Utils/Logger.h"
#include "Spark/Net/EventLoop.h"
#include "Spark/Net/Socket.h"

TcpClient::TcpClient(EventLoop* loop,
                     const SocketAddr& serverAddr,
                     std::string name)
    : loop_(loop),
      name_(std::move(name)),
      connector_(new Connector(loop_, serverAddr)),
      connectionCallback_(defaultConnectionCallback),
      messageCallback_(defaultMessageCallback),
      connect_(true)
{
    connector_->setNewConnectionCallback([this](int fd) { newConnection(fd); });
    LOG_TRACE << "TcpClient::TcpClient(" << name_ << ") constructed";
}

TcpClient::~TcpClient()
{
    LOG_TRACE << "~TcpClient(" << name_ << ")";
    TcpConnectionPtr conn;
    bool unique = false;
    {
        std::lock_guard<std::mutex> lg(mutex_);
        unique = connection_.unique();
        conn = connection_;
    }
    if (conn)
    {
        assert(loop_ == conn->getLoop());
        auto loop = loop_;
        loop_->runInLoop([conn, loop] {
            conn->setCloseCallback([loop](const TcpConnectionPtr& cp) {
                loop->queueInLoop([cp] { cp->connectDestroyed(); });
            });
        });
        if (unique)
            conn->forceClose();
        else
            LOG_TRACE << "conn [" << conn.get() << "] is not unique";
    }
    else
    {
        connector_->stop();
        LOG_TRACE << "conn is null, stop the connector_";
    }
}

void TcpClient::connect()
{
    LOG_TRACE << "TcpClient[" << name_ << "] connect to "
              << connector_->peerAddr().toStringPort();
    connect_ = true;
    connector_->start();
}

void TcpClient::disconnect()
{
    connect_.store(false);
    {
        std::lock_guard<std::mutex> lg(mutex_);
        if (connection_)
            connection_->shutdown();
    }
}

void TcpClient::stop()
{
    connect_.store(false);
    connector_->stop();
}

void TcpClient::newConnection(int sockfd)
{
    LOG_TRACE << "TcpClient::newConnection fd[" << sockfd << "]";
    assert(loop_->isInLoopThread());
    SocketAddr local = SocketAddr::getLocalAddr(sockfd);
    SocketAddr peer = SocketAddr::getPeerAddr(sockfd);
    char connName[128]{};
    snprintf(connName,
             sizeof(connName) - 1,
             "%s-%s#%d",
             name_.c_str(),
             peer.toStringPort().c_str(),
             nextConnId_.load());
    ++nextConnId_;
    auto conn =
        std::make_shared<TcpConnection>(loop_, sockfd, connName, local, peer);
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteDoneCallback(writeCompleteCallback_);
    conn->setCloseCallback(
        [this](const TcpConnectionPtr& conn) { removeConnection(conn); });
    {
        std::lock_guard<std::mutex> lg(mutex_);
        connection_ = conn;
    }
    conn->connectEstablished();
}

void TcpClient::removeConnection(const TcpConnectionPtr& conn)
{
    assert(loop_->isInLoopThread());
    assert(loop_ == conn->getLoop());
    {
        std::lock_guard<std::mutex> lg(mutex_);
        assert(connection_ == conn);
        connection_.reset();
    }

    loop_->queueInLoop([conn] { conn->connectDestroyed(); });

    if (retry_ && connect_)
    {
        LOG_TRACE << "TcpClient[" << name_ << "] reconnecting to "
                  << connector_->peerAddr().toStringPort();
        connector_->restart();
    }
}
